package com.cloud.mqtt.Emq;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.stereotype.Component;
import java.util.concurrent.TimeUnit;

/**
 * @program: mqtt
 * @description: mqtt服务端
 * @author: Cloud
 * @create: 2020/11/21 09:52:33
 */
@Slf4j
@Component
public class MQTTServer {

    public MqttClient client;

    public MqttTopic topic;

    public MqttMessage message;

    private static MQTTConnect mqttConnect = new MQTTConnect();

    public MQTTServer() throws MqttException {
        connect();
    }

    // 连接
    public void connect() throws MqttException {
        // 防止重复创建MQTTClient实例
        if (client == null) {
            client = new MqttClient(MqttConfig.pubHost + MqttConfig.pubPort,MqttConfig.pubClientId,new MemoryPersistence());
        }
        MqttConnectOptions options = mqttConnect.getOptions();
        options.setCleanSession(true);
        options.setUserName(MqttConfig.pubUserName);
        // 判断拦截状态，这里注意一下，如果没有这个判断，是非常坑的
        if (client.isConnected()) {
            // 连接成功之后重连!
            client.disconnect();
            client.connect(options);
            log.info("--------------------连接成功!-----------------------------");
        } else {
            client.connect(options);
            log.info("--------------------连接失败!尝试重新连接mqtt服务器...-----------------------------");
            if (client.isConnected()) {
                log.info("--------------------重连成功!-----------------------------");
            }
        }
    }

    // 发布
    public boolean publish(MqttTopic topic1, MqttMessage message) throws MqttException {
        MqttDeliveryToken token = topic1.publish(message);
        // 等待是否发送成功!
        token.waitForCompletion();
        log.info("message is published completely! "
                + token.isComplete());
        return token.isComplete();
    }

    // 发送消息
    public void sendMessage(String topic,String data) throws MqttException {
        log.info("本次转发的消息内容是：{}",data);
        log.info("本次转发的消息主题是：{}",topic);
        MQTTServer server = new MQTTServer();
        server.topic = server.client.getTopic(topic);
        server.message = new MqttMessage();
        // 如果重复消费，则把值改为true,然后发送一条空的消息，之前的消息就会覆盖，然后在改为false
        server.message.setRetained(false);
        server.message.setQos(1);
        server.message.setPayload(data.getBytes());
        server.publish(server.topic,server.message);
    }

    // 发布消息
    public void publishMessage(){
        while (true) {
            try {
                TimeUnit.SECONDS.sleep(100);
                //取出messageMap数据
                //发送数据
                sendMessage(MqttConfig.pubTopic,"hello,are you wendy?");
                log.info("本次发送的主题内容是：hello,are you wendy？");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (MqttException e) {
                e.printStackTrace();
            }

        }
    }
}
